package xyz.chaoslolomonica.recipes;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class DoubleDistributedBarrier implements PathChildrenCacheListener, Barrier {


	private static final String BARRIERNAME = "b";
	private final CuratorFramework client;
	private final ReentrantLock lock;
	private final Condition nodeChangeCondition;
	private final String rootPath;
	private final int waitOnCount;
	private PathChildrenCache nodeCache;

	public DoubleDistributedBarrier(CuratorFramework client, String rootPath, int waitOnCount) {
		this.client = client;
		lock = new ReentrantLock();
		this.nodeChangeCondition = lock.newCondition();
		this.rootPath = rootPath;
		this.waitOnCount = waitOnCount;
		try {
			Stat stat = this.client.checkExists().forPath(rootPath);
			if (stat == null) {
				this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(rootPath);
			}
			// register once
			//			this.client.getData().usingWatcher(new MyWatcher()).forPath(this.rootPath);
			this.nodeCache = new PathChildrenCache(client, rootPath, false);
			this.nodeCache.start();
			this.nodeCache.getListenable().addListener(this);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public String enter() {
		try {
			String realPath = this.client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
					.forPath(rootPath + "/" + BARRIERNAME);
			while (!Thread.interrupted()) {
				List<String> childrenList = client.getChildren().forPath(this.rootPath);
				if (childrenList.size() < this.waitOnCount) {
					lock.lock();
					try {
						System.out.println("enter condition is not satisfied and current size is " + childrenList.size());
						nodeChangeCondition.await(10, TimeUnit.MILLISECONDS);
					} finally {
						lock.unlock();
					}
				} else {
					return realPath;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public void leave(String barrierName) {
		try {
			this.client.delete().forPath(barrierName);
			while (!Thread.interrupted()) {
				List<String> childrenList = client.getChildren().forPath(this.rootPath);
				if (!childrenList.isEmpty()) {
					lock.lock();
					try {
						System.out.println("leave condition is not satisfied and current size is " + childrenList.size());
						nodeChangeCondition.await(10, TimeUnit.MILLISECONDS);
					} finally {
						lock.unlock();
					}
				} else {
					break;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent)
			throws Exception {
		lock.lock();
		try {
			System.out.println("node change event:" + pathChildrenCacheEvent.getType().name());
			this.nodeChangeCondition.signalAll();
		} finally {
			lock.unlock();
		}
	}
}
